package com.explorex.infra.puma.gateway.netty.ws;

import com.explorex.infra.puma.gateway.netty.common.*;
import com.explorex.infra.puma.gateway.netty.common.enums.EnumTerminal;
import com.explorex.infra.puma.gateway.netty.common.enums.EnumUserStatus;
import com.explorex.infra.puma.gateway.netty.common.security.AESUtil;
import com.explorex.infra.puma.gateway.netty.common.security.EDCHUtil;
import com.explorex.infra.puma.gateway.netty.common.utils.JacksonObjectMapper;
import com.explorex.infra.puma.gateway.netty.redis.BaseSessionRedis;
import com.explorex.infra.puma.gateway.netty.ws.enums.MessageTypeEnum;
import com.explorex.infra.puma.gateway.netty.ws.message.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.nio.channels.ClosedChannelException;
import java.security.KeyPair;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: JacceYang
 * @createTime: 2022/07/14 8:41 下午
 * @description: Channel Session information
 */
@Data
@Slf4j
public class WebSocketSession implements BaseSocketSession, BaseSessionAttribute {

    // 60 * 1000
    private static final int ONE_MINUTE = 60000;
    private final Channel channel;
    /**
     * user status 1- online ,2- offline ,3-background
     */
    private int status = EnumUserStatus.OFFLINE.getKey();

    private int preStatus = EnumUserStatus.OFFLINE.getKey();
    /**
     * 鉴权状态,初始化时为未鉴权状态
     */
    private AuthStatusEnum auth = AuthStatusEnum.NOT_AUTH;
    /**
     * 客户端(远端) Ip 地址
     */
    private String remoteIp;

    private int remotePort;
    /**
     * 本地机器链接Ip
     */
    private String localIp;
    /**
     * 网络连接 四元组
     */
    private String ipLink;

    /**
     * 设备信息
     */
    private Device device;

    /**
     * 用户信息
     */
    private String userId;
    /**
     * 用户所属的domain信息
     */
    private String appId;

    private AtomicInteger seqId = new AtomicInteger(0);

    /**
     * 密钥信息加密用
     */
    private byte[] shareKey;
    /**
     * 服务端密钥信息
     */
    private KeyPair keyPair;
    private int terminal = EnumTerminal.WEB.getValue();

    private UserKey authUser;

    /**
     * 是否熔断ack 重试逻辑，对于消息没有回复的ack 重试，极端场景关闭重试导致的client端的压力增大，例如弱网或者瞬间大流量，由于重试存在导致消息压力大。
     */
    private volatile boolean ackRetryFuse;

    private volatile long coldTime;

    private AtomicInteger failedTimes = new AtomicInteger(0);
    ;

    public AtomicInteger getFailedTimes() {
        return failedTimes;
    }

    public void setFailedTimes(int failedTimes) {
        this.failedTimes.set(failedTimes);
    }

    public boolean isAckRetryFuse() {
        return ackRetryFuse;
    }

    public void setAckRetryFuse(boolean ackRetryFuse) {
        // 首次 进入冷却
        if ((System.currentTimeMillis() - coldTime > 1000) && ackRetryFuse) {
            coldTime = System.currentTimeMillis();
        }
        this.ackRetryFuse = ackRetryFuse;
    }

    @Override
    public boolean isRateLimited() {
        return isAckRetryFuse() && System.currentTimeMillis() - coldTime < 1000;
    }

    @Override
    public UserKey getAuthUser() {
        return authUser;
    }

    public void setAuthUser(UserKey authUser) {
        this.authUser = authUser;
    }

    protected Map<String, UserKey> tokenUserInfo = new ConcurrentHashMap<>();

    @Override
    public Map<String, UserKey> getTokenUserInfo() {
        return tokenUserInfo;
    }

    /**
     * 鉴权的时候带过来的上下文信息
     */
    Map<String, String> extra;

    @Override
    public Map<String, String> getExtra() {
        return extra;
    }

    private volatile long lastTime = System.currentTimeMillis();
    private volatile long lastRefreshEventTime = System.currentTimeMillis();

    private static BaseSessionRedis baseSessionRedis = SpringContextUtils.getBean("baseSessionRedis", BaseSessionRedis.class);

    public WebSocketSession(Channel channel) {
        this.channel = channel;
    }

    public int nextSeqId() {
        return seqId.getAndAdd(1);
    }

    public Channel channel() {
        return channel;
    }

    @Override
    public String domain() {
        return appId;
    }

    @Override
    public String remoteIp() {
        return remoteIp;
    }

    @Override
    public int remotePort() {
        return remotePort;
    }

    public void setRemotePort(int remotePort) {
        this.remotePort = remotePort;
    }

    @Override
    public String localIp() {
        return localIp;
    }

    @Override
    public String ipLink() {
        return ipLink;
    }

    @Override
    public DeviceBase device() {
        return device;
    }

    @Override
    public String userId() {
        return userId;
    }

    public AuthStatusEnum getAuth() {
        return auth;
    }

    @Override
    public int userStatus() {
        return status;
    }

    @Override
    public int preStatus() {
        return preStatus;
    }

    @Override
    public void setStatus(int status) {
        this.preStatus = this.status;
        this.status = status;
    }

    @Override
    public int terminal() {
        return terminal;
    }

    @Override
    public int channelTag() {
        return 0;
    }

    @Override
    public boolean needLog() {
        return true;
    }

    @Override
    public ProtocolEnum protocol() {
        return ProtocolEnum.WS;
    }

    @Override
    public long getLastTime() {
        return lastTime;
    }

    @Override
    public void setLastTime(long lastTime) {
        this.lastTime = lastTime;
    }

    @Override
    public boolean canWrite() {
        return channel.isActive() && channel.isWritable();
    }

    public KeyPair keyPair() {
        if (keyPair == null) {
            keyPair = EDCHUtil.generateKeyPair();
        }
        return keyPair;
    }

    public void shareKey(byte[] shareKey) {
        return;
//        this.shareKey=shareKey;
//        this.channel.attr(ChannelConst.SHARE_KEY)
//                .set(String.valueOf(shareKey));
    }

    private ChannelFuture sendTextMessage(String message) {
        byte[] encrypt = null;
        try {
            encrypt = AESUtil.encrypt(shareKey, message.getBytes());
        } catch (Exception e) {
            log.error("sendTextMessage failed, message={}", message);
        }
        ChannelPromise channelPromise = channel.newPromise();
        byte[] finalEncrypt = encrypt;
        channel().eventLoop().execute(() -> channel.writeAndFlush(new TextWebSocketFrame(new String(finalEncrypt)), channelPromise));
        return channelPromise;
    }

    public void refresh(boolean force) {
        // 如果消息有上行，也可以更新心跳
        if (this.auth == AuthStatusEnum.AUTHED && (force || System.currentTimeMillis() - lastTime > 5000)) {
            lastTime = System.currentTimeMillis();
            this.baseSessionRedis.updateSessionInfo(this);
        }
    }

    private ChannelFuture sendBinaryMessage(ByteBuf byteBuf) {
        ChannelPromise channelPromise = channel.newPromise();
        channel().eventLoop().execute(() -> channel.writeAndFlush(new BinaryWebSocketFrame(byteBuf), channelPromise));
        return channelPromise;
    }

    @Override
    public ChannelFuture sendMessage(Object msg) {
        long startTime = System.currentTimeMillis();
        if (!canWrite()) {
            log.info("sendMessage unwritable session={}", this.toString());
            return channel.newSucceededFuture();
        }
        refresh(false);
        if (msg instanceof Message) {
            Message message = (Message) msg;
            return sendTextMessage(JacksonObjectMapper.dumpObjectUnchecked(message)).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    // todo：增加打点定位问题，后续可以去掉
                    if (message.getHeader().getType() == MessageTypeEnum.AUTH_ACK.getType() || message.getHeader().getType() == MessageTypeEnum.MESSAGE_ACK.getType()) {
                    }
                } else {
                    log.error("response sendMessage failed,session={},msg={},e={}", this, message, future.cause());

                    if (future.cause() instanceof ClosedChannelException) {
                        closeSocket("close");
                    }
                }
            });
        } else if (msg instanceof ByteBuf) {
            return sendBinaryMessage((ByteBuf) msg);
        }else if(msg instanceof String){
            return sendTextMessage((String) msg).addListener((ChannelFutureListener) future ->{
                if(future.isSuccess()){
                    log.info("isSuccess");
                }else {
                    log.warn("failed");
                }
            });
        }
        throw new UnsupportedOperationException();
    }

    @Override
    public ChannelFuture closeSocket(String msg) {
        channel().writeAndFlush(new CloseWebSocketFrame(3102, msg));
        log.info("server call close,session={},msg:{}", this, msg);
        return channel().close();
    }

    @Override
    public String toString() {
        return "WebSocketSession{" +
                "channel=" + channel +
                ", status=" + status +
                ", preStatus=" + preStatus +
                ", auth=" + auth +
                ", remoteIp='" + remoteIp + '\'' +
                ", ipLink='" + ipLink + '\'' +
                ", device=" + device +
                ", terminal=" + terminal +
                ", authUser=" + authUser +
                ", tokenUserInfo=" + tokenUserInfo +
                ", lastTime=" + lastTime +
                '}';
    }
}
